package com.navinfo.opentsp.platform.computing.analysis.service;

import com.navinfo.opentsp.platform.computing.analysis.entity.drivingAnalysis.VehicleInfo;
import com.navinfo.opentsp.platform.computing.analysis.entity.truckload.*;
import com.navinfo.opentsp.platform.computing.analysis.util.PropertiesUtil;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.*;
import scala.collection.JavaConverters;

import java.io.Serializable;
import java.math.BigDecimal;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.time.Instant;
import java.time.LocalDate;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 天级计算各车的坡度载荷油耗关系明细-专项针对2load,-4slope,10dw,42speed，存入HDFS
 */
public class TruckLoadStageDurationGear4DWService implements Serializable {
    /**
     * 档位计算需要的阀值如下
     */
    public static final int Gear_Speed_Threshold = 1;
    public static final int Gear_Rotation_Threshold = 500;
    public static final double Gear_Filter_Coefficient= 0.2;
    public static final int Gear_Threshold = 10000;
    public static final double Gear_Coefficient = 0.5;
    public static final int SECONDS_OF_DAY = 24 * 60 * 60;
    /**
     * 统计天数
     */
//    public static final int DAYS = 7;
    /**
     * 待统计carmodel
     */
//    public static final String CARMODEL = "Q141W33561T7942A9";
    /**
     * 统计时间
     */
    private Date curDate;

    /**
     * 调试标记
     */
    private boolean debug = false;
    /**
     *
     */
    private String hdfsPathHost;
    /**
     * spark SparkSession
     */
    private SparkSession spark;
    /**
     * 应用配置广播变量，用于算子内读取配置项
     */
    private Broadcast<Map<String, String>> configMap;

    /**
     * 初始化
     * @param spark spark
     * @param jsc JavaSparkContext
     * @param curDate 统计时间
     */
    public void init(SparkSession spark, JavaSparkContext jsc, Date curDate) {
        this.spark = spark;
        this.curDate = curDate;
        this.debug = Boolean.parseBoolean(PropertiesUtil.getProperties("monitor.debug")==null?
                "false":PropertiesUtil.getProperties("monitor.debug"));
        this.configMap = jsc.broadcast(PropertiesUtil.getProperties());
        this.hdfsPathHost = PropertiesUtil.getProperties("hdfs.path.host");

    }

    /**
     * 执行统计方法
     * @param jsc JavaSparkContext
     */
    public void run(JavaSparkContext jsc) {
        try {
            //开始日期转换成Date类型 YYYY-MM-DD
            ZoneId zoneId = ZoneId.systemDefault();
            Instant startInstant = curDate.toInstant();

            LocalDate startLocalDate = startInstant.atZone(zoneId).toLocalDate();

            //获取各车载荷最小最大载荷HDFS数据
            Dataset<Row>   metricsDS = getMetricsRDD(jsc).filter("car_model='"+configMap.getValue().get("truckload.slope.carmodel")+"'");
            if (debug) {
                System.out.println("startLocalDate:"+startLocalDate.toString());
                System.out.println(metricsDS.count());
                metricsDS.show();
                metricsDS.printSchema();
            }

            //from mysql，加载车辆档位转数等信息
            final Broadcast<Map<Long, VehicleInfo>> vehicleInfoMap = initVehicleCache(jsc, configMap);

            for (int i = 0; i < Integer.parseInt(configMap.getValue().get("truckload.slope.days")); i++) {
                calcOneday(jsc,startLocalDate.minusDays(i),metricsDS,vehicleInfoMap);
                System.out.println("statday==========="+changeLocalDate2String(startLocalDate.minusDays(i)));
            }

            spark.read().parquet(hdfsPathHost+"/truckload/tid_slope_dw_day_4").createOrReplaceTempView("spark_inner_tempview_ts");

            Dataset<Row> tidSlopeTLDS = spark.sql("select tid,carModel,loadStage,slope,concat_ws(':',collect_list(point)) as points from " +
                    "(select tid,carModel,loadStage,slope,concat_ws(',',dw,speed,fuelConsumption,totalMilage," +
                    "startGpstime,endGpstime,startLatitude,endLatitude,startLongitude,endLongitude) as point from " +
                    "(select tid,carModel,loadStage,slope,dw,speed,totalFuel*100/totalMilage as  fuelConsumption,totalMilage," +
                    "startGpstime,endGpstime,startLatitude,endLatitude,startLongitude,endLongitude " +
                    "from spark_inner_tempview_ts  ) ts) " +
                    " where  " +
                    "loadStage=2 and  slope=-4 and dw=10 and speed=42 " +
                    "group by tid,carModel,loadStage,slope");

            //每段数据计算出最小油耗的对应数据
            Dataset<TidSlope> tidSlopeDSResult = tidSlopeTLDS.flatMap(new FlatMapFunction<Row, TidSlope>() {
                    @Override
                    public Iterator<TidSlope> call(Row row) throws Exception {
                        List<TidSlope> result = new ArrayList<>();
                        long tid = row.getLong(0);
                        String carModel = row.getString(1);
                        int loadStage = row.getInt(2);
                        int slope = row.getInt(3);
                        String points = row.getString(4);
                        float minTotalFuel = Float.MAX_VALUE;
                        int minDW = 0;
                        int minSpeed = 0;
                        float slopeMilage = 0;
                        int startGpstime = 0;
                        int endGpstime = 0;
                        int startLatitude = 0;
                        int endLatitude = 0;
                        int startLongitude = 0;
                        int endLongitude = 0;

                        String[] a = points.split(":");
                        for (int i = 0; i < a.length; i++) {
                            String[] b = a[i].split(",");
                            minTotalFuel= Float.parseFloat(b[2]);
                            minDW = Integer.parseInt(b[0]);
                            minSpeed= Integer.parseInt(b[1]);
                            slopeMilage = Float.parseFloat(b[3]);
                            startGpstime = Integer.parseInt(b[4]);
                            endGpstime = Integer.parseInt(b[5]);
                            startLatitude = Integer.parseInt(b[6]);
                            endLatitude = Integer.parseInt(b[7]);
                            startLongitude = Integer.parseInt(b[8]);
                            endLongitude = Integer.parseInt(b[9]);
                            result.add(TidSlope.builder().tid(tid).carModel(carModel).dw(minDW).
                                    loadStage(loadStage).slope(slope).totalFuel(minTotalFuel).speed(minSpeed).
                                    slopeMilage(slopeMilage).
                                    startGpstime(startGpstime).startLatitude(startLatitude).startLongitude(startLongitude).
                                    endGpstime(endGpstime).endLatitude(endLatitude).endLongitude(endLongitude).build());
                        }

                        return result.iterator();
                    }},Encoders.bean(TidSlope.class));

            String yearMonthDay = changeLocalDate2String(startLocalDate);

            //结果存入HDFS
           tidSlopeDSResult.repartition(1).write().option("header", "true").
                   format("csv").mode(SaveMode.Overwrite).save(hdfsPathHost+"/truckload/tid_slope_dw_4/part_time="+yearMonthDay);

        } catch(Exception e){
            e.printStackTrace();
        }

    }

    private void calcOneday(JavaSparkContext jsc, LocalDate startLocalDate,Dataset<Row>   metricsDS,Broadcast<Map<Long, VehicleInfo>> vehicleInfoMap) {
        // 获取HDFS行程表前天数据
        JavaRDD<Row> curRdd = getTripRDD(jsc, startLocalDate.minusDays(2),1);
//        if (debug) {
//            curRdd.foreach(doc -> System.out.println(doc.toJson()));
//        }

        //将JavaRDD<Document>转化为JavaPairRDD<Long, TruckLoadEstimated>
        JavaRDD<TruckLoadEstimated> curTLERdd = convertRdd(curRdd, curDate);
        Dataset<Row> curTLEDS = spark.createDataFrame(curTLERdd, TruckLoadEstimated.class);
        if (debug) {
            System.out.println(curTLEDS.count());
            curTLEDS.show();
            curTLEDS.printSchema();
        }

        //curTLEDS左关联metricsDS
        Dataset<Row> curMetricsDS = curTLEDS.join(metricsDS,
                JavaConverters.asScalaIteratorConverter(Arrays.asList("terminalId").iterator()).asScala().toSeq(),
                "left_outer")
                .selectExpr(("terminalId,vehicleestimatedload,routeEndTime,routeStartTime," +
                        "levelRoadDrivingFuelConsumption,levelRoadDrivingMileage,engineSpeedRange03FuelConsumption," +
                        "engineSpeedRange03Mileage,engineSpeedRange02FuelConsumption,engineSpeedRange02Mileage," +
                        "engineSpeedRange01FuelConsumption,engineSpeedRange01Mileage," +
                        "empty_load,half_load,full_load,car_model," +
                        "routeIntegralMileage,routeIntegralFuelConsumption," +
                        "routeStartLatitude,routeStartLongitude,routeStartHeight," +
                        "routeEndLatitude,routeEndLongitude,routeEndHeight").split(","))
                .where("empty_load is not null");
        if (debug) {
            System.out.println(curMetricsDS.count());
            curMetricsDS.show();
            curMetricsDS.printSchema();
        }

        //计算载荷档位，及其各档时长、油耗、里程
        MapFunction<Row, TruckLoadDuration> basicMapFunc = row -> {
            TruckLoadDuration tld = computeDurationByLoad(row);
            return tld;
        };

        //计算载荷档位，及其各档时长、油耗、里程
        Dataset<TruckLoadDuration> tldDS = curMetricsDS.map(basicMapFunc,Encoders.bean(TruckLoadDuration.class));
        if (debug) {
            System.out.println(tldDS.count());
            tldDS.show();
            tldDS.printSchema();
        }

        tldDS.createOrReplaceTempView("spark_inner_tempview_tl");

        //https://spark.apache.org/docs/2.3.0/api/sql/index.html
        Dataset<Row> tripTLDS = spark.sql("select terminalId,concat_ws(':',collect_list(tripLoad)) as triploads from (select terminalId,concat_ws(',',routeStartTime,routeEndTime,loadStage,carModel) as tripLoad from spark_inner_tempview_tl) tl group by terminalId");

        Dataset<Row> f37TLDS = spark.sql("select terminalId,concat_ws(':',collect_list(point)) as points from " +
                "(select tid as terminalId,concat_ws(',',gpstime,height,engineoutputtorque/100,speed/100,accelerator/100,rotation/100,fuelconsumptionrate/100,brake,clutchswitch,latitude,longitude) as point " +
                "from navinfo_parquet_0f37 where part_time=" + changeLocalDate2String(startLocalDate.minusDays(2))+
                ") f37 group by terminalId");

        if (debug) {
            tripTLDS.show(false);
            tripTLDS.printSchema();
            f37TLDS.show(false);
            f37TLDS.printSchema();
        }

        //0F37数据和行程载荷档位数据关联
        Dataset<Row> f37LoadDS = f37TLDS.join(tripTLDS,
                JavaConverters.asScalaIteratorConverter(Arrays.asList("terminalId").iterator()).asScala().toSeq(),
                "inner");

        if (debug) {
            f37LoadDS.show(false);
            f37LoadDS.printSchema();
        }

        //转换为车型、坡度、档位等100个点一段的数据
        Dataset<TidSlope> tidSlopeDS = f37LoadDS.flatMap(new FlatMapFunction<Row, TidSlope>() {
            @Override
            public Iterator<TidSlope> call(Row row) throws Exception {
                long tid = Long.parseLong(row.getString(0));
                String points = row.getString(1);
                String triploads = row.getString(2);
                List<F37Point> f37Points = new ArrayList<>();
                List<TripLoad> tripLoads = new ArrayList<>();
                List<TidSlope> result = new ArrayList<>();

                String[] a= points.split(":");
                for(int i=0;i<a.length;i++){
                    String[] b= a[i].split(",");
                    String[] b_last=new String[10];
                    int filter_height = 0;
                    float filter_speed = 0;
                    if (i>0) {
                        b_last= a[i-1].split(",");
                    }
                    //fix bug @ 20181226141900
                    if ((i>0)&&((Integer.parseInt(String.valueOf(Long.parseLong(b[0])/1000))-Integer.parseInt(String.valueOf(Long.parseLong(b_last[0])/1000)))==1)) {
                        //针对相邻2点的高程和速度做低通滤波
                        double height_cutoff = 0.3;
                        double height_alpha = 0.1/(1/(2.0*3.1415926*height_cutoff) + 0.1);
                        filter_height = (int)((height_alpha*(Integer.parseInt(b[1])) + (1-height_alpha)*Integer.parseInt(b_last[1]))+0.5);
                        double speed_cutoff = 10;
                        double speed_alpha = 0.1/(1/(2.0*3.1415926*speed_cutoff) + 0.1);
                        filter_speed = (float)((speed_alpha*(Float.parseFloat(b[3])) + (1-speed_alpha)*Float.parseFloat(b_last[3]))+0.5);
                    } else {
                        filter_height = Integer.parseInt(b[1]);
                        filter_speed = Float.parseFloat(b[3]);
                    }

                    f37Points.add(
                            F37Point.builder().tid(tid).gpstime(Integer.parseInt(String.valueOf(Long.parseLong(b[0])/1000))).
                                    height(filter_height).
                                    engineoutputtorque(Float.parseFloat(b[2])).
                                    speed(filter_speed).
                                    accelerator(Float.parseFloat(b[4])).
                                    rotation(Float.parseFloat(b[5])).
                                    fuelconsumptionrate(Float.parseFloat(b[6])).
                                    brake(Integer.parseInt(b[7])).
                                    clutchswitch(Integer.parseInt(b[8])).
                                    latitude(Integer.parseInt(b[9])).
                                    longitude(Integer.parseInt(b[10])).build()
                    );
                }

                a= triploads.split(":");
                for(int i=0;i<a.length;i++){
                    String[] b= a[i].split(",");
                    tripLoads.add(
                            TripLoad.builder().routeStartTime(Integer.parseInt(b[0])).carModel(b[3]).
                                    routeEndTime(Integer.parseInt(b[1])).loadStage(Integer.parseInt(b[2])).build()
                    );
                }
                tripLoads.sort(Comparator.comparingInt(TripLoad::getRouteStartTime));

                for(TripLoad vb : tripLoads){
//                    System.out.println(vb.getRouteStartTime());
                    int loadStage = vb.getLoadStage();
                    String carModel = vb.getCarModel();
                    // 存放过滤结果的列表
                    List<F37Point> F37Point4Tripload = null;
                    // 使用lambda表达式过滤出结果并放到result列表里
                    F37Point4Tripload = f37Points.stream().filter((F37Point fpp) -> ((fpp.getGpstime()>=vb.getRouteStartTime())
                            &&(fpp.getGpstime()<=vb.getRouteEndTime()))).collect(Collectors.toList());
                    F37Point4Tripload.sort(Comparator.comparingInt(F37Point::getGpstime));
                    int count = 0;
                    int slope = -1;
                    int speed =0;
                    F37Point firstF37Point = null;
                    F37Point secondF37Point = null;
                    F37Point lastF37Point = null;
                    int upOrDownFlag = 0;
                    float totalFuel = 0;
                    float totalMilage = 0;
                    double totalDuration = 0;
                    int dw = 0;
                    for(F37Point fpp : F37Point4Tripload){
                        if (count==0) {
                            firstF37Point = fpp;
                        }
                        if (count==1) {
                            secondF37Point = fpp;
                        }


//                        if (count==2) {
////                                System.out.println(firstF37Point.toString());
////                                System.out.println(secondF37Point.toString());
////                                slope = Math.abs(Math.round((secondF37Point.getHeight()-firstF37Point.getHeight())/((firstF37Point.getSpeed()*1000/3600)*
////                                        ((secondF37Point.getGpstime()-firstF37Point.getGpstime())))*100));
//                            double v = Math.atan2(secondF37Point.getHeight()-firstF37Point.getHeight(), ((firstF37Point.getSpeed()*1000/3600)*
//                                    ((secondF37Point.getGpstime()-firstF37Point.getGpstime()))));
//                            slope = (int) Math.round(v * 180 / Math.PI);
//                        }

                        if (lastF37Point!=null) {
                            totalFuel= totalFuel+(fpp.getFuelconsumptionrate()*(fpp.getGpstime()-lastF37Point.getGpstime()))/(3600);
                            totalMilage= totalMilage+(fpp.getSpeed()*(fpp.getGpstime()-lastF37Point.getGpstime()))/(3600);
                            totalDuration = totalDuration+(fpp.getGpstime()-lastF37Point.getGpstime());
                        }

                        dw = calcDW(tid, vehicleInfoMap, firstF37Point) + 1;
//                            System.out.println(firstF37Point.toString());
//                            System.out.println(dw);

                        if (count == 1) {
                            upOrDownFlag = (fpp.getHeight() - lastF37Point.getHeight()) >= 0 ? 1 : -1;
                        }
                        if (count > 1) {
                            int upOrDown = (fpp.getHeight() - lastF37Point.getHeight()) >= 0 ? 1 : -1;
                            if (upOrDownFlag * upOrDown < 0) {
                                count = 0;
                                slope = -1;
                                speed =0;
                                firstF37Point = null;
                                secondF37Point = null;
                                lastF37Point = null;
                                totalFuel = 0;
                                totalMilage = 0;
                                totalDuration = 0;
                                dw = 0;
                                continue;
                            }
                        }
                        lastF37Point = fpp;
                        count++;

                        if ((fpp.getBrake()==100)||(fpp.getClutchswitch()==100)) {
                            //点不能包含刹车和离合切换，否则重新初始化
                            count = 0;
                            slope = -1;
                            speed =0;
                            firstF37Point = null;
                            secondF37Point = null;
                            lastF37Point = null;
                            totalFuel = 0;
                            totalMilage = 0;
                            totalDuration = 0;
                            dw = 0;
                            continue;
                        }

//                                System.out.println("totalMilage:"+totalMilage);
//                                System.out.println("totalDuration:"+totalDuration);
                        //速度按照2km为区间划分
                        speed =  (((int)((totalMilage*3600/totalDuration+0.5)/2))*2+((int)(totalMilage*3600/totalDuration+0.5)%2)*2);
                        //坡度按照百分比表示法
                        slope = Math.round(((fpp.getHeight()-firstF37Point.getHeight())/ (totalMilage*1000))*100);
                        if ((dw>0) &&((totalMilage*1000)>=100)&&(Math.abs(slope)<15)) {//点里程差大于100m
                            result.add(TidSlope.builder().tid(tid).carModel(carModel).
                                        loadStage(loadStage).slope(slope).dw(dw).speed(speed).
                                        totalFuel(totalFuel).totalMilage(totalMilage).endGpstime(fpp.getGpstime()).
                                        startGpstime(firstF37Point.getGpstime()).
                                        startLatitude(firstF37Point.getLatitude()).endLatitude(fpp.getLatitude()).
                                        startLongitude(firstF37Point.getLongitude()).endLongitude(fpp.getLongitude()).build());
                            //重新初始化
                            count = 0;
                            slope = -1;
                            speed =0;
                            firstF37Point = null;
                            secondF37Point = null;
                            lastF37Point = null;
                            totalFuel = 0;
                            totalMilage = 0;
                            totalDuration = 0;
                            dw = 0;
                        }
                    }
                }
                return result.iterator();
            }
        },Encoders.bean(TidSlope.class));

        if (debug) {
            tidSlopeDS.show(false);
            tidSlopeDS.printSchema();
        }

        String yearMonthDay = changeLocalDate2String(startLocalDate.minusDays(2));

        //结果存入HDFS
        tidSlopeDS.write().format("parquet").
                mode(SaveMode.Overwrite).save(hdfsPathHost+"/truckload/tid_slope_dw_day_4/part_time="+yearMonthDay);
    }

    /**
     * 档位计算
     * @param tid 终端号
     * @param vehicleInfoMap 车辆配置信息
     * @param dataBean 0f37数据点
     * @return 档位
     */
    private static int calcDW(long tid, Broadcast<Map<Long, VehicleInfo>> vehicleInfoMap, F37Point dataBean) {
        VehicleInfo vehicleInfo = vehicleInfoMap.getValue().get(tid);
//        logger.debug("===============档位算法开始===========");
        if (vehicleInfo == null) {
//            logger.debug("挡位算法：获取车辆信息为空，返回");
            return -2;
        }

        double[] gearsSpeedRatio = vehicleInfo.getGearsSpeedRatio();
        if (gearsSpeedRatio == null || gearsSpeedRatio.length == 0) {
//            logger.debug("挡位算法：变速箱速比为null，返回");
            return -2;
        }

        //如果速度小于阈值不计算
        if (dataBean.getSpeed()/3.6 <= Gear_Speed_Threshold ) {
            return -2;
        }
        float rotation = dataBean.getRotation();
        float speedKmh = dataBean.getSpeed();
        double accelerator = dataBean.getAccelerator();

        int index = -2;
        //如果rotation<=750&&油门开度=0，计为空挡，统计到其它挡位中
        if (!(rotation <= Gear_Rotation_Threshold && accelerator == 0)) {
            double rearAxleSpeedRatio = vehicleInfo.getRearAxleSpeedRatio();
            double wheelRadius = vehicleInfo.getWheelRadius();
            double m = rotation * 2 * Math.PI * wheelRadius * 60;

            double n = speedKmh * rearAxleSpeedRatio * 1000;

            if (n == 0) {
                return -2;
            }
            //计算档位
            double gear = BigDecimal.valueOf(m)
                    .divide(BigDecimal.valueOf(n), 2, BigDecimal.ROUND_HALF_UP)
                    .doubleValue();
            double filterGear = gear;

            double min = Gear_Threshold;
            for (int i = 0; i < gearsSpeedRatio.length; i++) {
                double delta = Math.abs(gearsSpeedRatio[i] - filterGear);
                if (delta < min) {
                    index = i;
                    min = delta;
                }
            }

            double diffGearsSpeedRatio = Math.abs(gearsSpeedRatio[index] - filterGear);
            //如果计算的速比与最接近的变速箱速比差值大于阈值，记为空挡
            if (BigDecimal.valueOf(diffGearsSpeedRatio).compareTo(BigDecimal.valueOf(Gear_Coefficient)) > 0) {
                index = -1;
            }
        }
        return index;

    }

    /**
     * 获取车辆相关信息：档位转数比等等
     * @param jsc spark
     * @param configMap mysql连接配置
     * @return 车辆相关信息：档位转数比等等
     * @throws SQLException  SQLException
     * @throws ClassNotFoundException ClassNotFoundException
     */
    private static Broadcast<Map<Long, VehicleInfo>> initVehicleCache(JavaSparkContext jsc, Broadcast<Map<String, String>> configMap) throws SQLException, ClassNotFoundException {
        String url = configMap.getValue().get("mysql.url");
        // MySQL配置时的用户名
        String user = configMap.getValue().get("mysql.username");
        // MySQL配置时的密码
        String password = configMap.getValue().get("mysql.password");
        // sql 查询语句
        String sql = configMap.getValue().get("vehicle.sql");

        HashMap<Long, VehicleInfo> vehicleInfoMap = new HashMap<>();
        Class.forName("com.mysql.jdbc.Driver");
        try (
                java.sql.Connection conn = DriverManager.getConnection(url, user, password);
                Statement statement = conn.createStatement();
                ResultSet rs = statement.executeQuery(sql);
        ) {
            while (rs.next()) {
                Long terminalId = rs.getLong("terminal_id");
                double rearaxleRatio = rs.getDouble("rearaxle_ratio");
                String gearRatios = rs.getString("gear_ratios");
                double wheelRadius = rs.getDouble("wheel_radius");
                double maxTorque = rs.getDouble("max_torque");
                double[] gearsSpeedRatio = null;
                if(gearRatios != null){
                    String[] gearRatiosArr = gearRatios.split(",");
                    gearsSpeedRatio = new double[gearRatiosArr.length];
                    for(int i=0; i<gearRatiosArr.length; i++){
                        gearsSpeedRatio[i]= Double.parseDouble(gearRatiosArr[i]);
                    }
                }

                VehicleInfo vehicleInfo = VehicleInfo.builder()
                        .terminalId(terminalId)
                        .rearAxleSpeedRatio(rearaxleRatio)
                        .wheelRadius(wheelRadius)
                        .gearsSpeedRatio(gearsSpeedRatio)
                        .engineMaxTorque(maxTorque)
                        .build();

                vehicleInfoMap.put(terminalId, vehicleInfo);
            }
        }
        return jsc.broadcast(vehicleInfoMap );
    }


    /**
     * 通过配置净载荷和最大载荷，计算载荷档位时长
     * @param row 车辆数据
     * @return 载荷档位时长对象
     */
    private TruckLoadDuration computeDurationByLoad(Row row) {
        TruckLoadDuration tld = new TruckLoadDuration();
        long terminalId = row.getAs("terminalId");
        int vehicleestimatedload = row.getAs("vehicleestimatedload");
        int routeEndTime = row.getAs("routeEndTime");
        int routeStartTime = row.getAs("routeStartTime");
        int mileage = row.getAs("routeIntegralMileage");
        int fuelconsumption = row.getAs("routeIntegralFuelConsumption");
//        "routeStartLatitude","routeStartLongitude","routeStartHeight",
//                "routeEndLatitude","routeEndLongitude","routeEndHeight");
        int routeStartLatitude = row.getAs("routeStartLatitude");
        int routeStartLongitude = row.getAs("routeStartLongitude");
        int routeStartHeight = row.getAs("routeStartHeight");
        int routeEndLatitude = row.getAs("routeEndLatitude");
        int routeEndLongitude = row.getAs("routeEndLongitude");
        int routeEndHeight = row.getAs("routeEndHeight");
        int empty_load = row.getAs("empty_load");
        int half_load = row.getAs("half_load");
        int full_load = row.getAs("full_load");
        String car_model = row.getAs("car_model");

        tld.setTerminalId(terminalId);
        tld.setCarModel(car_model);
        tld.setRouteStartTime(routeStartTime);
        tld.setRouteEndTime(routeEndTime);
        tld.setRouteIntegralMileage(mileage);
        tld.setRouteIntegralFuelConsumption(fuelconsumption);
        tld.setRouteStartLatitude(routeStartLatitude);
        tld.setRouteStartLongitude(routeStartLongitude);
        tld.setRouteStartHeight(routeStartHeight);
        tld.setRouteEndLatitude(routeEndLatitude);
        tld.setRouteEndLongitude(routeEndLongitude);
        tld.setRouteEndHeight(routeEndHeight);
        //载荷档位，枚举0：空载，1：半载，2：满载，3：超载
        if (vehicleestimatedload<empty_load) {
            tld.setEmptyLoadDuration(routeEndTime-routeStartTime);
            tld.setEmptyLoadMileage(mileage);
            tld.setEmptyLoadFuelconsumption(fuelconsumption);
            tld.setLoadStage(0);
        } else if ((vehicleestimatedload>=empty_load)&&(vehicleestimatedload<half_load)) {
            tld.setHalfLoadDuration(routeEndTime-routeStartTime);
            tld.setHalfLoadMileage(mileage);
            tld.setHalfLoadFuelconsumption(fuelconsumption);
            tld.setLoadStage(1);
        } else if ((vehicleestimatedload>=half_load)&&(vehicleestimatedload<full_load)) {
            tld.setFullLoadDuration(routeEndTime-routeStartTime);
            tld.setFullLoadMileage(mileage);
            tld.setFullLoadFuelconsumption(fuelconsumption);
            tld.setLoadStage(2);
        } else if (vehicleestimatedload>=full_load) {
            tld.setExtraLoadDuration(routeEndTime-routeStartTime);
            tld.setExtraLoadMileage(mileage);
            tld.setExtraLoadFuelconsumption(fuelconsumption);
            tld.setLoadStage(3);
        }
        return tld;
    }

    /**
     * 类型转换
     * @param ld LocalDate类型数据
     * @return String类型数据
     */
    private String changeLocalDate2String(LocalDate ld) {
        DateTimeFormatter fmt = DateTimeFormatter.ofPattern("yyyyMMdd");
        return ld.format(fmt);
    }

    /**
     * 将JavaRDD<Document>JavaRDD<TruckLoadEstimated>
     * @param rdd Document的RDD数据
     * @param date 统计日期
     * @return JavaRDD<TruckLoadEstimated>格式的RDD数据
     */
    private JavaRDD<TruckLoadEstimated> convertRdd(JavaRDD<Row> rdd, Date date) {
        return rdd.map(new Function<Row, TruckLoadEstimated>() {
            @Override
            public TruckLoadEstimated call(Row doc) throws Exception {
                TruckLoadEstimated rt = new TruckLoadEstimated();
                rt.setTimeEstimated(date.getTime());
//                Dataset<Row> result = spark.sql("select terminalId,vehicleEstimatedLoad,routeStartTime," +
//                "routeEndTime,levelRoadDrivingFuelConsumption,levelRoadDrivingMileage," +
//                "engineSpeedRange03FuelConsumption,engineSpeedRange03Mileage," +
//                "engineSpeedRange02FuelConsumption,engineSpeedRange02Mileage," +
//                "engineSpeedRange01FuelConsumption,engineSpeedRange01Mileage," +
//                "routeIntegralMileage,routeIntegralFuelConsumption," +
//                "routeStartLatitude,routeStartLongitude,routeStartHeight," +
//                "routeEndLatitude,routeEndLongitude,routeEndHeight" +
//                " from sdata");
                // 终端唯一标识
                rt.setTerminalId(Long.parseLong(String.valueOf(doc.get(0))));
                // 估计载荷
                rt.setVehicleestimatedload(Integer.parseInt(String.valueOf(doc.get(1))));
                rt.setRouteEndTime(Integer.parseInt(String.valueOf(doc.get(3))));
                rt.setRouteStartTime(Integer.parseInt(String.valueOf(doc.get(2))));
                rt.setLevelRoadDrivingFuelConsumption(Integer.parseInt(String.valueOf(doc.get(4))));
                rt.setLevelRoadDrivingMileage(Integer.parseInt(String.valueOf(doc.get(5))));
                rt.setEngineSpeedRange03FuelConsumption(Integer.parseInt(String.valueOf(doc.get(6))));
                rt.setEngineSpeedRange03Mileage(Integer.parseInt(String.valueOf(doc.get(7))));
                rt.setEngineSpeedRange02FuelConsumption(Integer.parseInt(String.valueOf(doc.get(8))));
                rt.setEngineSpeedRange02Mileage(Integer.parseInt(String.valueOf(doc.get(9))));
                rt.setEngineSpeedRange01FuelConsumption(Integer.parseInt(String.valueOf(doc.get(10))));
                rt.setEngineSpeedRange01Mileage(Integer.parseInt(String.valueOf(doc.get(11))));
                rt.setRouteIntegralMileage(Integer.parseInt(String.valueOf(doc.get(12))));
                rt.setRouteIntegralFuelConsumption(Integer.parseInt(String.valueOf(doc.get(13))));
                rt.setRouteStartLatitude(Integer.parseInt(String.valueOf(doc.get(14))));
                rt.setRouteStartLongitude(Integer.parseInt(String.valueOf(doc.get(15))));
                rt.setRouteStartHeight(Integer.parseInt(String.valueOf(doc.get(16))));
                rt.setRouteEndLatitude(Integer.parseInt(String.valueOf(doc.get(17))));
                rt.setRouteEndLongitude(Integer.parseInt(String.valueOf(doc.get(18))));
                rt.setRouteEndHeight(Integer.parseInt(String.valueOf(doc.get(19))));
                return rt;
            }
        });
    }

    private Dataset<Row> getMetricsRDD(JavaSparkContext jsc) {
        Dataset<Row> resultDF = spark.read().option("header","true").
                option("inferSchema","true").csv(hdfsPathHost+"/truckload/tid_truckload_car_type.csv").
                select("terminalId","empty_load","half_load","full_load","car_model").toDF();

        return resultDF;

    }

    /**
     * 获取某表所在HDFS行程表数据
     * @param jsc JavaSparkContext
     * @param curDate 统计日期
     * @param days 天数
     * @return 某表所在周的mongodb行程表数据
     */
    private JavaRDD<Row> getTripRDD(JavaSparkContext jsc, LocalDate curDate, int days) {
        String statDay = changeLocalDate2String(curDate.minusDays(0));
        if (debug) {
            System.out.println("statDay:"+statDay);
        }

        Dataset<Row> statisticDataDF = spark.read().parquet(hdfsPathHost+"/tripanalysis/trip/"+statDay.substring(0,4)+"/"+
                statDay.substring(4,6)+"/"+statDay.substring(6,8)+"/data/*");

        for (int i = 1; i < days; i++) {
            statDay = changeLocalDate2String(curDate.minusDays(i));
            if (debug) {
                System.out.println("statDay:"+statDay);
            }

            Dataset<Row> statisticDataDFOneday = spark.read().parquet(hdfsPathHost+"/tripanalysis/trip/"+statDay.substring(0,4)+"/"+
                    statDay.substring(4,6)+"/"+statDay.substring(6,8)+"/data/*");
            statisticDataDF = statisticDataDF.union(statisticDataDFOneday);
        }

        /**
         * 注册成为临时表以供后续的SQL查询操作
         */
        statisticDataDF.registerTempTable("sdata");
        /**
         * 进行数据的多维度分析
         */
        Dataset<Row> result = spark.sql("select terminalId,vehicleEstimatedLoad,routeStartTime," +
                "routeEndTime,levelRoadDrivingFuelConsumption,levelRoadDrivingMileage," +
                "engineSpeedRange03FuelConsumption,engineSpeedRange03Mileage," +
                "engineSpeedRange02FuelConsumption,engineSpeedRange02Mileage," +
                "engineSpeedRange01FuelConsumption,engineSpeedRange01Mileage," +
                "routeIntegralMileage,routeIntegralFuelConsumption," +
                "routeStartLatitude,routeStartLongitude,routeStartHeight," +
                "routeEndLatitude,routeEndLongitude,routeEndHeight" +
                " from sdata where vehicleEstimatedLoad>0");

        if (debug) {
            System.out.println(result.count());
            result.show();
            result.printSchema();
        }
        return result.javaRDD();
    }
}
